#include "ORCL.h"

// 创建第i个normal rc
void orcl_create_normal_conn_qps(struct orcl_context *orcl_ctx, int i)
{
    assert(orcl_ctx->rc_normal_qp[i] == NULL &&
           orcl_ctx->rc_normal_cq[i] == NULL && orcl_ctx->rc_normal_pd[i] != NULL && orcl_ctx->rc_normal_ctx[i] != NULL);
    assert(orcl_ctx->dev_port_id >= 1);

    // 首先创造一个完成队列
    orcl_ctx->rc_normal_cq[i] = ibv_create_cq(orcl_ctx->rc_normal_ctx[i], ORCL_SQ_DEPTH, NULL, NULL, 0);
    assert(orcl_ctx->rc_normal_cq[i] != NULL);

    struct ibv_qp_init_attr create_attr;
    memset(&create_attr, 0, sizeof(struct ibv_qp_init_attr));
    create_attr.send_cq = orcl_ctx->rc_normal_cq[i];
    create_attr.recv_cq = orcl_ctx->rc_normal_cq[i];

    create_attr.qp_type = IBV_QPT_RC;
    create_attr.cap.max_send_wr = ORCL_SQ_DEPTH;
    // 因为要为基于连接的方式使用send_recv，所以recv队列也要设计一定的长度
    create_attr.cap.max_recv_wr = ORCL_SQ_DEPTH;
    create_attr.cap.max_send_sge = 10;
    create_attr.cap.max_recv_sge = 10;
    create_attr.cap.max_inline_data = ORCL_RC_MAX_INLINE;
    create_attr.sq_sig_all = 0;

    orcl_ctx->rc_normal_qp[i] = ibv_create_qp(orcl_ctx->rc_normal_pd[i], &create_attr);
    assert(orcl_ctx->rc_normal_qp[i] != NULL);

    struct ibv_qp_attr init_attr;
    memset(&init_attr, 0, sizeof(struct ibv_qp_attr));
    init_attr.qp_state = IBV_QPS_INIT;
    init_attr.pkey_index = 0;
    init_attr.port_num = orcl_ctx->dev_port_id;
    init_attr.qp_access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ |
                                IBV_ACCESS_REMOTE_ATOMIC;

    if (ibv_modify_qp(orcl_ctx->rc_normal_qp[i], &init_attr,
                      IBV_QP_STATE |
                          IBV_QP_PKEY_INDEX |
                          IBV_QP_PORT |
                          IBV_QP_ACCESS_FLAGS))
    {
        fprintf(stderr, "Failed to modify conn QP to INIT\n");
        exit(-1);
    }
}

void orcl_create_forward_conn_qps(struct orcl_context *orcl_ctx, int i)
{
    assert(orcl_ctx->rc_forward_qp[i] == NULL &&
           orcl_ctx->rc_forward_cq[i] == NULL && orcl_ctx->rc_forward_pd[i] != NULL && orcl_ctx->rc_forward_ctx[i] != NULL);
    assert(orcl_ctx->dev_port_id >= 1);

    // 首先创造一个完成队列
    orcl_ctx->rc_forward_cq[i] = ibv_create_cq(orcl_ctx->rc_forward_ctx[i], ORCL_SQ_DEPTH, NULL, NULL, 0);
    assert(orcl_ctx->rc_forward_cq[i] != NULL);

    struct ibv_qp_init_attr create_attr;
    memset(&create_attr, 0, sizeof(struct ibv_qp_init_attr));
    create_attr.send_cq = orcl_ctx->rc_forward_cq[i];
    create_attr.recv_cq = orcl_ctx->rc_forward_cq[i];

    create_attr.qp_type = IBV_QPT_RC;
    create_attr.cap.max_send_wr = ORCL_SQ_DEPTH;
    // 因为要为基于连接的方式使用send_recv，所以recv队列也要设计一定的长度
    create_attr.cap.max_recv_wr = ORCL_SQ_DEPTH;
    create_attr.cap.max_send_sge = 10;
    create_attr.cap.max_recv_sge = 10;
    create_attr.cap.max_inline_data = ORCL_RC_MAX_INLINE;
    create_attr.sq_sig_all = 0;

    orcl_ctx->rc_forward_qp[i] = ibv_create_qp(orcl_ctx->rc_forward_pd[i], &create_attr);
    assert(orcl_ctx->rc_forward_qp[i] != NULL);

    struct ibv_qp_attr init_attr;
    memset(&init_attr, 0, sizeof(struct ibv_qp_attr));
    init_attr.qp_state = IBV_QPS_INIT;
    init_attr.pkey_index = 0;
    init_attr.port_num = orcl_ctx->dev_port_id;
    init_attr.qp_access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ |
                                IBV_ACCESS_REMOTE_ATOMIC;

    if (ibv_modify_qp(orcl_ctx->rc_forward_qp[i], &init_attr,
                      IBV_QP_STATE |
                          IBV_QP_PKEY_INDEX |
                          IBV_QP_PORT |
                          IBV_QP_ACCESS_FLAGS))
    {
        fprintf(stderr, "Failed to modify forward conn QP to INIT\n");
        exit(-1);
    }
}

void orcl_normal_qp_connect(struct orcl_context *ctx, int i, struct orcl_qp_attr *remote_qp_attr)
{
    assert(ctx != NULL);
    assert(ctx->rc_normal_qp[i] != NULL && ctx->rc_normal_pd[i] != NULL &&
           ctx->rc_normal_cq[i] != NULL && ctx->rc_normal_ctx[i] != NULL);

    assert(remote_qp_attr != NULL);
    // normal rc的数量小于group的size
    assert(i < ctx->group_size);
    assert(ctx->dev_port_id >= 1);

    // 记录远程节点的基本信息
    memcpy(&(ctx->others_normal_qp[i]), remote_qp_attr, sizeof(struct orcl_qp_attr));

    struct ibv_qp_attr conn_attr;
    memset(&conn_attr, 0, sizeof(struct ibv_qp_attr));
    conn_attr.qp_state = IBV_QPS_RTR;
    conn_attr.path_mtu = IBV_MTU_4096;
    conn_attr.dest_qp_num = remote_qp_attr->qpn;
    conn_attr.rq_psn = ORCL_DEFAULT_PSN;

    conn_attr.ah_attr.is_global = 0;
    conn_attr.ah_attr.dlid = remote_qp_attr->lid;
    conn_attr.ah_attr.sl = 0;
    conn_attr.ah_attr.src_path_bits = 0;
    conn_attr.ah_attr.port_num = ctx->dev_port_id; /* Local port! */

#if USE_GID_ROUTE == 1
    conn_attr.ah_attr.is_global = 1;
    conn_attr.ah_attr.grh.dgid = remote_qp_attr->gid;
    conn_attr.ah_attr.grh.sgid_index = 0;
    conn_attr.ah_attr.grh.hop_limit = 128;
#endif

    int rtr_flags = IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU |
                    IBV_QP_DEST_QPN | IBV_QP_RQ_PSN;

    conn_attr.max_dest_rd_atomic = 16;
    conn_attr.min_rnr_timer = 12;
    rtr_flags |= IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER;

    // 进入rtr，
    if (ibv_modify_qp(ctx->rc_normal_qp[i], &conn_attr, rtr_flags))
    {
        fprintf(stderr, "ORCL: Failed to modify QP to RTR\n");
        assert(false);
    }

    memset(&conn_attr, 0, sizeof(conn_attr));
    conn_attr.qp_state = IBV_QPS_RTS;
    conn_attr.sq_psn = ORCL_DEFAULT_PSN;

    int rts_flags = IBV_QP_STATE | IBV_QP_SQ_PSN;

    conn_attr.timeout = 14;
    conn_attr.retry_cnt = 7;
    conn_attr.rnr_retry = 7;
    conn_attr.max_rd_atomic = 16;
    conn_attr.max_dest_rd_atomic = 16;
    rts_flags |= IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY |
                 IBV_QP_MAX_QP_RD_ATOMIC;

    if (ibv_modify_qp(ctx->rc_normal_qp[i], &conn_attr, rts_flags))
    {
        fprintf(stderr, "ORCL: Failed to modify QP to RTS\n");
        assert(false);
    }

    return;
}

void orcl_forward_qp_connect(struct orcl_context *ctx, int i, struct orcl_qp_attr *remote_qp_attr)
{
    assert(ctx != NULL);
    assert(ctx->rc_forward_qp[i] != NULL && ctx->rc_forward_pd[i] != NULL &&
           ctx->rc_forward_cq[i] != NULL && ctx->rc_forward_ctx[i] != NULL);

    assert(remote_qp_attr != NULL);
    // normal rc的数量小于group的size
    assert(i < ctx->group_size);
    assert(ctx->dev_port_id >= 1);

    memcpy(&(ctx->others_forward_qp[i]), remote_qp_attr, sizeof(struct orcl_qp_attr));

    struct ibv_qp_attr conn_attr;
    memset(&conn_attr, 0, sizeof(struct ibv_qp_attr));
    conn_attr.qp_state = IBV_QPS_RTR;
    conn_attr.path_mtu = IBV_MTU_4096;
    conn_attr.dest_qp_num = remote_qp_attr->qpn;
    conn_attr.rq_psn = ORCL_DEFAULT_PSN;

    conn_attr.ah_attr.is_global = 0;
    conn_attr.ah_attr.dlid = remote_qp_attr->lid;
    conn_attr.ah_attr.sl = 0;
    conn_attr.ah_attr.src_path_bits = 0;
    conn_attr.ah_attr.port_num = ctx->dev_port_id; /* Local port! */

#if USE_GID_ROUTE == 1
    conn_attr.ah_attr.is_global = 1;
    conn_attr.ah_attr.grh.dgid = remote_qp_attr->gid;
    conn_attr.ah_attr.grh.sgid_index = 0;
    conn_attr.ah_attr.grh.hop_limit = 128;
#endif

    int rtr_flags = IBV_QP_STATE | IBV_QP_AV | IBV_QP_PATH_MTU |
                    IBV_QP_DEST_QPN | IBV_QP_RQ_PSN;

    conn_attr.max_dest_rd_atomic = 16;
    conn_attr.min_rnr_timer = 12;
    rtr_flags |= IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER;

    // 进入rtr
    if (ibv_modify_qp(ctx->rc_forward_qp[i], &conn_attr, rtr_flags))
    {
        fprintf(stderr, "ORCL: Failed to modify QP to RTR\n");
        assert(false);
    }

    memset(&conn_attr, 0, sizeof(conn_attr));
    conn_attr.qp_state = IBV_QPS_RTS;
    conn_attr.sq_psn = ORCL_DEFAULT_PSN;

    int rts_flags = IBV_QP_STATE | IBV_QP_SQ_PSN;

    conn_attr.timeout = 14;
    conn_attr.retry_cnt = 7;
    conn_attr.rnr_retry = 7;
    conn_attr.max_rd_atomic = 16;
    conn_attr.max_dest_rd_atomic = 16;
    rts_flags |= IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT | IBV_QP_RNR_RETRY |
                 IBV_QP_MAX_QP_RD_ATOMIC;

    // 进入rts
    if (ibv_modify_qp(ctx->rc_forward_qp[i], &conn_attr, rts_flags))
    {
        fprintf(stderr, "ORCL: Failed to modify QP to RTS\n");
        assert(false);
    }

    return;
}

// 所有的RC节点建立连接
void orcl_all_qp_connect(struct orcl_context *ctx)
{
    // 在组内所有的节点与之前的节点建立连接

    // 遍历组内所有节点，遍历自己的所有的
    int i;
    for (i = 0; i < ctx->group_size; i++)
    {
        // 不需要与自己建立连接
        int inner_group_index = ctx->node_id % ctx->group_size;

        if(inner_group_index == i){
            continue;
        }

        // 所有的rc qp要和远程节点的对应rc建立连接
        // "server-normal-%d-%d-%d", ctx->node_id, ctx->orcl_ctx_id, i
        // 第一个是节点号，第二个是节点内上下文好，第三个是其RC要连接的远程节点
        // 在这里的实现node_id就是当前迭代变量i与groupsize相关的一个变量
        // ，ctx_id和当前节点保持一致
        // 最后一个是当前节点的本组连接好
        // 获取对应RC的内容
        char remote_qp_name[ORCL_QP_NAME_SIZE];

        int remote_node_id = ctx->group_size * ctx->group_id + i;
        
        sprintf(remote_qp_name, "server-normal-%d-%d-%d", remote_node_id, ctx->orcl_ctx_id, inner_group_index);

        // 获取对应远程节点的qp信息
        struct orcl_qp_attr *remote_qp = NULL;

        printf("find info of:%s\n", remote_qp_name);

        while (remote_qp == NULL)
        {
            remote_qp = orcl_get_published_qp(remote_qp_name);
            if (remote_qp == NULL)
            {
                usleep(200000);
            }
        }

        // 连接远程的qp
        orcl_normal_qp_connect(ctx, i, remote_qp);

        // 连接forward，连接方式和normal一样
        memset(remote_qp_name, 0, ORCL_QP_NAME_SIZE);
        
        sprintf(remote_qp_name, "server-forward-%d-%d-%d", remote_node_id, ctx->orcl_ctx_id, inner_group_index);

        remote_qp = NULL;

        while (remote_qp == NULL)
        {
            remote_qp = orcl_get_published_qp(remote_qp_name);
            if (remote_qp == NULL)
            {
                usleep(200000);
            }
        }

        orcl_forward_qp_connect(ctx, i, remote_qp);
    }
}

// 发布一个变量来进行跨进程同步
void orcl_publish_ready(const char *key_str){
    char value[ORCL_QP_NAME_SIZE];
	assert(key_str != NULL && strlen(key_str) < ORCL_QP_NAME_SIZE);

	char new_name[2 * ORCL_QP_NAME_SIZE];
	sprintf(new_name, "%s", ORCL_RESERVED_NAME_PREFIX);
	strcat(new_name, key_str);
	
	sprintf(value, "%s", "orcl_ready");
	orcl_publish(new_name, value, strlen(value));
}

void orcl_wait_till_ready(const char *key_str){
    char *value;
	char exp_value[ORCL_QP_NAME_SIZE];
	sprintf(exp_value, "%s", "orcl_ready");
	
	char new_name[2 * ORCL_QP_NAME_SIZE];
	sprintf(new_name, "%s", ORCL_RESERVED_NAME_PREFIX);
	strcat(new_name, key_str);

	int tries = 0;
	while(true) {
		int ret = orcl_get_published(new_name, (void **) &value);
		tries++;
		if(ret > 0) {
			if(strcmp(value, exp_value) == 0) {
				return;
			}
		}

		usleep(200000);

		if(tries > 100) {
			fprintf(stderr, "ORCL: Waiting for key word %s to be ready\n", key_str);
			tries = 0;
		}
	}
}

void sync_create_dc_initiators(struct orcl_context_sync* sync_ctx, int i){
    assert(sync_ctx != NULL);
    assert(sync_ctx->dci_ctx[i] != NULL);
    assert(sync_ctx->dev_port_id >= 1);

    sync_ctx->dci_cq[i] = ibv_create_cq(sync_ctx->dci_ctx[i], ORCL_SQ_DEPTH, NULL, NULL, 0);
    assert(sync_ctx->dci_cq[i] != NULL);

    // 创建DCT initiator，也就是发送队列
    struct ibv_qp_init_attr_ex create_attr;
    memset((void *) &create_attr, 0, sizeof(create_attr));
    // 发送和接收使用同一个完成队列，但是接收完成应该是没有用的，因为initiator不接收
    create_attr.send_cq = sync_ctx->dci_cq[i];
    create_attr.recv_cq = sync_ctx->dci_cq[i];

    // 这里没有最大接收队列的参数
    create_attr.cap.max_send_wr = ORCL_SQ_DEPTH;
    create_attr.cap.max_send_sge = 10;
    create_attr.cap.max_inline_data = ORCL_DCT_MAX_INLINE;
    create_attr.qp_type = IBV_EXP_QPT_DC_INI;
    create_attr.pd = sync_ctx->dci_pd[i];
    // 这个是DC和XRC会使用的
    create_attr.comp_mask = IBV_QP_INIT_ATTR_PD;

    // 这里的qp创建的和DCT有关qp
    errno = 0;

    sync_ctx->dci_qp[i] = ibv_create_qp_ex(sync_ctx->dci_ctx[i], &create_attr);
    // 判断错误
    if(errno == EINVAL){
        printf("Invalid pd, send_cq, recv_cq, srq or invalid value provided in max_send_wr, max_recv_wr, max_send_sge, max_recv_sge or in max_inline_data\n");
    }else if(errno == ENOMEM){
        printf("Not enough resources to complete this operation\n");
    }else if(errno == ENOSYS){
        printf("QP with this Transport Service Type isn't supported by this RDMA device\n");
    }else if(errno == EPERM){
        printf("Not enough permissions to create a QP with this Transport Service Type\n");
    }
    
    assert(sync_ctx->dci_qp[i] != NULL);

    // 进入init状态
    struct ibv_exp_qp_attr modify_attr;
    memset((void *) &modify_attr, 0, sizeof(modify_attr));
    modify_attr.qp_state = IBV_QPS_INIT;
    modify_attr.pkey_index = 0;
    modify_attr.port_num = sync_ctx->dev_port_id;
    // 在RC的模式中主要是IBV_ACCESS_REMOTE_WRITE等有关于qp权限的内容
    modify_attr.qp_access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ |
            IBV_ACCESS_REMOTE_ATOMIC;
    modify_attr.dct_key = ORCL_DCT_KEY;

    // 进入初始化状态
    if (ibv_exp_modify_qp(sync_ctx->dci_qp[i], &modify_attr,
        IBV_EXP_QP_STATE | IBV_EXP_QP_PKEY_INDEX |
        IBV_EXP_QP_PORT | IBV_EXP_QP_DC_KEY)) {
        fprintf(stderr, "Failed to modify QP to INIT\n");
        assert(false);
    }

    // 这个函数从这里开始需要远程节点的信息一开始都和自己连接
    /* Modify QP to RTR */
    // 这里到达ready to receive状态
    modify_attr.qp_state = IBV_QPS_RTR;
    modify_attr.max_dest_rd_atomic = 0;
    modify_attr.path_mtu = IBV_MTU_4096;

    // 和远程节点连接，在UD中没有这么复杂的参数，这里更像是RC的参数配置
    modify_attr.ah_attr.is_global = 0;
    /* Initially, connect to the DCT target on machine 0 */
    // 一开始统一连接到自己身上，hrd_get_local_lid用这个接口
    modify_attr.ah_attr.dlid = orcl_get_local_lid(sync_ctx->dci_ctx[i], sync_ctx->dev_port_id);
    // 用本地的哪一个端口去连接
    modify_attr.ah_attr.port_num = sync_ctx->dev_port_id;	/* Local port */
    modify_attr.ah_attr.sl = 0;
    modify_attr.dct_key = ORCL_DCT_KEY;

    // 进入RTR状态
    if (ibv_exp_modify_qp(sync_ctx->dci_qp[i], &modify_attr, IBV_EXP_QP_STATE |
        //IBV_EXP_QP_MAX_DEST_RD_ATOMIC | IBV_EXP_QP_PATH_MTU |
        IBV_EXP_QP_PATH_MTU |
        IBV_EXP_QP_AV)) {
        fprintf(stderr, "Failed to modify QP to RTR\n");
        assert(false);
    }

    /* Modify QP to RTS */
    // 进入RTS
    modify_attr.qp_state = IBV_QPS_RTS;
    modify_attr.timeout = 14;
    modify_attr.retry_cnt = 7;
    modify_attr.rnr_retry = 7;
    modify_attr.max_rd_atomic = 16;
    if (ibv_exp_modify_qp(sync_ctx->dci_qp[i], &modify_attr,
        IBV_EXP_QP_STATE | IBV_EXP_QP_TIMEOUT | IBV_EXP_QP_RETRY_CNT |
        IBV_EXP_QP_RNR_RETRY | IBV_EXP_QP_MAX_QP_RD_ATOMIC)) {
        fprintf(stderr, "Failed to modify QP to RTS\n");
        assert(false);
    }
}


// 为上下文创建第i个dci
void orcl_create_dc_initiators(struct orcl_context* orcl_ctx, int i){
    assert(orcl_ctx != NULL);
    assert(orcl_ctx->dci_ctx[i] != NULL);
    assert(orcl_ctx->dev_port_id >= 1);

    orcl_ctx->dci_cq[i] = ibv_create_cq(orcl_ctx->dci_ctx[i], ORCL_SQ_DEPTH, NULL, NULL, 0);
    assert(orcl_ctx->dci_cq[i] != NULL);

    // 创建DCT initiator，也就是发送队列
    struct ibv_qp_init_attr_ex create_attr;
    memset((void *) &create_attr, 0, sizeof(create_attr));
    // 发送和接收使用同一个完成队列，但是接收完成应该是没有用的，因为initiator不接收
    create_attr.send_cq = orcl_ctx->dci_cq[i];
    create_attr.recv_cq = orcl_ctx->dci_cq[i];

    // 这里没有最大接收队列的参数
    create_attr.cap.max_send_wr = ORCL_SQ_DEPTH;
    create_attr.cap.max_send_sge = 10;
    create_attr.cap.max_inline_data = ORCL_DCT_MAX_INLINE;
    create_attr.qp_type = IBV_EXP_QPT_DC_INI;
    create_attr.pd = orcl_ctx->dci_pd[i];
    // 这个是DC和XRC会使用的
    create_attr.comp_mask = IBV_QP_INIT_ATTR_PD;

    // 这里的qp创建的和DCT有关qp
    errno = 0;

    orcl_ctx->dci_qp[i] = ibv_create_qp_ex(orcl_ctx->dci_ctx[i], &create_attr);
    // 判断错误
    if(errno == EINVAL){
        printf("Invalid pd, send_cq, recv_cq, srq or invalid value provided in max_send_wr, max_recv_wr, max_send_sge, max_recv_sge or in max_inline_data\n");
    }else if(errno == ENOMEM){
        printf("Not enough resources to complete this operation\n");
    }else if(errno == ENOSYS){
        printf("QP with this Transport Service Type isn't supported by this RDMA device\n");
    }else if(errno == EPERM){
        printf("Not enough permissions to create a QP with this Transport Service Type\n");
    }
    
    assert(orcl_ctx->dci_qp[i] != NULL);

    // 进入init状态
    struct ibv_exp_qp_attr modify_attr;
    memset((void *) &modify_attr, 0, sizeof(modify_attr));
    modify_attr.qp_state = IBV_QPS_INIT;
    modify_attr.pkey_index = 0;
    modify_attr.port_num = orcl_ctx->dev_port_id;
    // 在RC的模式中主要是IBV_ACCESS_REMOTE_WRITE等有关于qp权限的内容
    modify_attr.qp_access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ |
            IBV_ACCESS_REMOTE_ATOMIC;
    modify_attr.dct_key = ORCL_DCT_KEY;

    // 进入初始化状态
    if (ibv_exp_modify_qp(orcl_ctx->dci_qp[i], &modify_attr,
        IBV_EXP_QP_STATE | IBV_EXP_QP_PKEY_INDEX |
        IBV_EXP_QP_PORT | IBV_EXP_QP_DC_KEY)) {
        fprintf(stderr, "Failed to modify QP to INIT\n");
        assert(false);
    }

    // 这个函数从这里开始需要远程节点的信息一开始都和自己连接
    /* Modify QP to RTR */
    // 这里到达ready to receive状态
    modify_attr.qp_state = IBV_QPS_RTR;
    modify_attr.max_dest_rd_atomic = 0;
    modify_attr.path_mtu = IBV_MTU_4096;

    // 和远程节点连接，在UD中没有这么复杂的参数，这里更像是RC的参数配置
    modify_attr.ah_attr.is_global = 0;
    /* Initially, connect to the DCT target on machine 0 */
    // 一开始统一连接到自己身上，hrd_get_local_lid用这个接口
    modify_attr.ah_attr.dlid = orcl_get_local_lid(orcl_ctx->dci_ctx[i], orcl_ctx->dev_port_id);
    // 用本地的哪一个端口去连接
    modify_attr.ah_attr.port_num = orcl_ctx->dev_port_id;	/* Local port */
    modify_attr.ah_attr.sl = 0;
    modify_attr.dct_key = ORCL_DCT_KEY;

    // 进入RTR状态
    if (ibv_exp_modify_qp(orcl_ctx->dci_qp[i], &modify_attr, IBV_EXP_QP_STATE |
        //IBV_EXP_QP_MAX_DEST_RD_ATOMIC | IBV_EXP_QP_PATH_MTU |
        IBV_EXP_QP_PATH_MTU |
        IBV_EXP_QP_AV)) {
        fprintf(stderr, "Failed to modify QP to RTR\n");
        assert(false);
    }

    /* Modify QP to RTS */
    // 进入RTS
    modify_attr.qp_state = IBV_QPS_RTS;
    modify_attr.timeout = 14;
    modify_attr.retry_cnt = 7;
    modify_attr.rnr_retry = 7;
    modify_attr.max_rd_atomic = 16;
    if (ibv_exp_modify_qp(orcl_ctx->dci_qp[i], &modify_attr,
        IBV_EXP_QP_STATE | IBV_EXP_QP_TIMEOUT | IBV_EXP_QP_RETRY_CNT |
        IBV_EXP_QP_RNR_RETRY | IBV_EXP_QP_MAX_QP_RD_ATOMIC)) {
        fprintf(stderr, "Failed to modify QP to RTS\n");
        assert(false);
    }

}

void sync_create_dc_target(struct orcl_context_sync* sync_ctx){
    assert(sync_ctx != NULL);
    assert(sync_ctx->dct_ctx != NULL && sync_ctx->dct_pd != NULL
        && sync_ctx->dct_mes_mr != NULL && sync_ctx->message_buffer != NULL);
    
    check_dct_supported(sync_ctx->dct_ctx);

    /* Create CQ. We don't use it at the target. */
	struct ibv_cq *cq = ibv_create_cq(sync_ctx->dct_ctx, ORCL_RQ_DEPTH, NULL, NULL, 0);
	assert(cq != NULL);
	// 赋值
    sync_ctx->dc_target_cq = cq;

    /* Create SRQ. SRQ is required to init DCT target, but we don't used it */
	struct ibv_srq_init_attr attr;
	memset((void *) &attr, 0, sizeof(attr));
	attr.attr.max_wr = 100;
	attr.attr.max_sge = 1;
	struct ibv_srq *srq = ibv_create_srq(sync_ctx->dct_pd, &attr);
	assert(srq != NULL);
	// 赋值
	sync_ctx->dc_target_srq = srq;

    // 开始创造target
	struct ibv_exp_dct_init_attr dctattr;
	memset((void *) &dctattr, 0, sizeof(dctattr));
	// dct target
	dctattr.pd = sync_ctx->dct_pd;
	// dct是和cq绑定的
	dctattr.cq = cq;
	// 接收队列，share的意思应该是这个接收队列会收到不同发送端发来的包
	dctattr.srq = srq;
	dctattr.dc_key = ORCL_DCT_KEY;
	dctattr.port = sync_ctx->dev_port_id;
	dctattr.access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ;
	dctattr.min_rnr_timer = 2;
	dctattr.tclass = 0;
	dctattr.flow_label = 0;
	dctattr.mtu = IBV_MTU_4096;
	dctattr.pkey_index = 0;
	dctattr.hop_limit = 1;

	dctattr.create_flags = 0;
	dctattr.inline_size = 150;

	// 返回的数据结构中有是一个DC target
	sync_ctx->dc_target = ibv_exp_create_dct(sync_ctx->dct_ctx, &dctattr);
	assert(sync_ctx->dc_target != NULL);

	check_dct_healthy(sync_ctx->dc_target);
}

void orcl_create_dc_target(struct orcl_context* orcl_ctx, int i){
    assert(orcl_ctx != NULL);
    assert(orcl_ctx->dct_ctx[i] != NULL && orcl_ctx->dct_pd[i] != NULL
        && orcl_ctx->dct_forward_mr[i] != NULL && orcl_ctx->dct_forward_buf[i]!=NULL);
    
    check_dct_supported(orcl_ctx->dct_ctx[i]);

    /* Create CQ. We don't use it at the target. */
	struct ibv_cq *cq = ibv_create_cq(orcl_ctx->dct_ctx[i], ORCL_RQ_DEPTH, NULL, NULL, 0);
	assert(cq != NULL);
	// 赋值
    orcl_ctx->dc_target_cq[i] = cq;

    /* Create SRQ. SRQ is required to init DCT target, but we don't used it */
	struct ibv_srq_init_attr attr;
	memset((void *) &attr, 0, sizeof(attr));
	attr.attr.max_wr = 100;
	attr.attr.max_sge = 1;
	struct ibv_srq *srq = ibv_create_srq(orcl_ctx->dct_pd[i], &attr);
	assert(srq != NULL);
	// 赋值
	orcl_ctx->dc_target_srq[i] = srq;

    // 开始创造target
	struct ibv_exp_dct_init_attr dctattr;
	memset((void *) &dctattr, 0, sizeof(dctattr));
	// dct target
	dctattr.pd = orcl_ctx->dct_pd[i];
	// dct是和cq绑定的
	dctattr.cq = cq;
	// 接收队列，share的意思应该是这个接收队列会收到不同发送端发来的包
	dctattr.srq = srq;
	dctattr.dc_key = ORCL_DCT_KEY;
	dctattr.port = orcl_ctx->dev_port_id;
	dctattr.access_flags = IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_REMOTE_READ;
	dctattr.min_rnr_timer = 2;
	dctattr.tclass = 0;
	dctattr.flow_label = 0;
	dctattr.mtu = IBV_MTU_4096;
	dctattr.pkey_index = 0;
	dctattr.hop_limit = 1;

	dctattr.create_flags = 0;
	dctattr.inline_size = 150;

	// 返回的数据结构中有是一个DC target
	orcl_ctx->dc_target[i] = ibv_exp_create_dct(orcl_ctx->dct_ctx[i], &dctattr);
	assert(orcl_ctx->dc_target[i] != NULL);

	check_dct_healthy(orcl_ctx->dc_target[i]);
}